﻿// See https://aka.ms/new-console-template for more information
using Confluent.Kafka;
using CSRedis;
using Newtonsoft.Json;
using QskjLibrary;
using QskjServiceKafka.Dto;
using System.Data;

Console.WriteLine("Hello, World!");
#region 消费数据
Task.Factory.StartNew(() =>
{
    string brokerList = "192.168.3.52:9092,192.168.3.146:9092,192.168.3.164:9092";
    List<string> topics = new List<string>();
    topics.Add("DEVICE_STATUS");
    Console.WriteLine(DateTime.Now.ToString() + "qskj设备状态GPS告警数据kafka消费开始1");

    //Thread.Sleep(1000 * 20);
    try
    {
        Consumer(brokerList, topics);
    }
    catch (Exception e)
    {
        Console.WriteLine("数据消费异常" + e);
    }
    while (true)
    {
        Thread.Sleep(1000 * 200);
    }
});
Task.Factory.StartNew(() =>
{
    string brokerList = "192.168.3.52:9092,192.168.3.146:9092,192.168.3.164:9092";
    List<string> topics = new List<string>();
    topics.Add("VEHICLE_STATUS");
    Console.WriteLine(DateTime.Now.ToString() + "qskj设备状态GPS告警数据kafka消费开始2");

    //Thread.Sleep(1000 * 20);
    try
    {
        Consumer(brokerList, topics);
    }
    catch (Exception e)
    {
        Console.WriteLine("数据消费异常" + e);
    }
    while (true)
    {
        Thread.Sleep(1000 * 200);
    }
});
Task.Factory.StartNew(() =>
{
    string brokerList = "192.168.3.52:9092,192.168.3.146:9092,192.168.3.164:9092";
    List<string> topics = new List<string>();
    topics.Add("VEHICLE_GPS");
    Console.WriteLine(DateTime.Now.ToString() + "qskj设备状态GPS告警数据kafka消费开始3");

    //Thread.Sleep(1000 * 20);
    try
    {
        Consumer(brokerList, topics);
    }
    catch (Exception e)
    {
        Console.WriteLine("数据消费异常" + e);
    }
    while (true)
    {
        Thread.Sleep(1000 * 200);
    }
});
Task.Factory.StartNew(() =>
{
    string brokerList = "192.168.3.52:9092,192.168.3.146:9092,192.168.3.164:9092";
    List<string> topics = new List<string>();
    topics.Add("VEHICLE_ALARM");
    Console.WriteLine(DateTime.Now.ToString() + "qskj设备状态GPS告警数据kafka消费开始4");

    //Thread.Sleep(1000 * 20);
    try
    {
        Consumer(brokerList, topics);
    }
    catch (Exception e)
    {
        Console.WriteLine("数据消费异常" + e);
    }
    while (true)
    {
        Thread.Sleep(1000 * 200);
    }
});
Task.Factory.StartNew(() =>
{
    string brokerList = "192.168.3.52:9092,192.168.3.146:9092,192.168.3.164:9092";
    List<string> topics = new List<string>();
    topics.Add("VEHICLE_ALARM_FILE");
    Console.WriteLine(DateTime.Now.ToString() + "qskj设备状态GPS告警数据kafka消费开始5");

    //Thread.Sleep(1000 * 20);
    try
    {
        Consumer(brokerList, topics);
    }
    catch (Exception e)
    {
        Console.WriteLine("数据消费异常" + e);
    }
    while (true)
    {
        Thread.Sleep(1000 * 200);
    }
});
#endregion

while (true)
{
    Thread.Sleep(5000);
}
void DeviceStatusDataToRedis(DeviceStatus status, CSRedisClient redisDb, CSRedisClient redisDb4)
{
    try
    {
        if (status.accessPlatformNode != "172.17.0.1")
        {
            string zl = redisDb.Get($@"zl_{status.devGbId}");
            if (!string.IsNullOrWhiteSpace(zl) && (status.platform == "RUIMING" || status.platform == "ZAKJ"))
            {
                return;
            }
            string dev = redisDb4.Get($@"lastonlinetime:{status.devGbId}");
            if (Convert.ToDateTime(dev) < DateTime.Today && status.devStatus.ToLower() == "online")//当天第一次上线
            {
                Console.WriteLine("lastonlinetime:" + dev);
                string sql = $@"insert into qskjrecord.D_VehicleOnlineData(vehicleNo,datetimed)values('{status.devGbId}','{DateTime.Now.ToString("yyyy-MM-dd")}')";
                MySqlHelper.ExecuteNonQuery(CommandType.Text, sql);
            }
            redisDb.Set(status.devGbId, (status.devStatus.ToLower().Contains("online") ? 0 : 1), new TimeSpan(0, 2, 1));//0-在线，1-离线
            redisDb4.Set(status.devGbId, status.devStatus.ToLower(), new TimeSpan(0, 2, 1));//online-在线，offline-离线
            if (status.devStatus.ToLower().Contains("online"))
            {
                redisDb4.Set($@"lastonlinetime:{status.devGbId}", DateTime.Now.ToString("yyyy-MM-dd HH:mm:ss"));//online-最后在线时间
            }
            else
            {
                redisDb4.Set($@"lastofflinetime:{status.devGbId}", DateTime.Now.ToString("yyyy-MM-dd HH:mm:ss"));//online-最后离线时间
            }
        }
    }
    catch (Exception e)
    {
        Console.WriteLine("设备状态信息写入Redis异常：" + JsonConvert.SerializeObject(status) + JsonConvert.SerializeObject(e));
    }
}
void VehicleGpsDataToRedis(VehicleGps status, CSRedisClient redisDb, CSRedisClient redisDb2)
{
    string zl = redisDb.Get($@"zl_{status.devGbId}");
    if (!string.IsNullOrWhiteSpace(zl) && (status.platform == "RUIMING" || status.platform == "ZAKJ"))
    {
        return;
    }
    int carSta = (((Convert.ToInt32(status.carState) & 8) == 8) ? 1 : 0);
    try
    {
        if (status.latitude == "0" || status.longitude == "0" || string.IsNullOrWhiteSpace(status.latitude) || string.IsNullOrWhiteSpace(status.longitude))
        {
            redisDb.Set("GPS0" + status.devGbId, status.latitude + "_" + status.longitude + "_" + carSta + "_" + status.carState + "_" + status.platform + "_" + DateTime.Now.ToString("yyyy-MM-dd HH:mm:ss"));
            redisDb2.Set("GPS0" + status.devGbId, status.latitude + "_" + status.longitude + "_" + carSta + "_" + status.carState + "_" + status.platform + "_" + DateTime.Now.ToString("yyyy-MM-dd HH:mm:ss"));
        }
        else
        {
            redisDb.Set("GPS" + status.devGbId, status.latitude + "_" + status.longitude + "_" + carSta + "_" + status.carState + "_" + status.platform + "_" + DateTime.Now.ToString("yyyy-MM-dd HH:mm:ss"));
            redisDb2.Set("GPS" + status.devGbId, status.latitude + "_" + status.longitude + "_" + carSta + "_" + status.carState + "_" + status.platform + "_" + DateTime.Now.ToString("yyyy-MM-dd HH:mm:ss"));
        }
    }
    catch (Exception e)
    {
        Console.WriteLine("设备GPS信息状态信息写入Redis异常：" + JsonConvert.SerializeObject(status) + JsonConvert.SerializeObject(e));
    }
}
void VehicleAlarmDataToRedis(VehicleAlarm status, CSRedisClient redisDb3, string aramlInfo, CSRedisClient redisDb)
{
    try
    {
        string zl = redisDb.Get($@"zl_{status.devGbId}");
        if (!string.IsNullOrWhiteSpace(zl) && (status.platform == "RUIMING" || status.platform == "ZAKJ"))
        {
            return;
        }
        redisDb3.Set("alarm" + status.devGbId, aramlInfo);
        redisDb3.Set(status.alarmId, aramlInfo, new TimeSpan(12, 0, 0));//报警事件存入Redis，有效期12小时
    }
    catch (Exception e)
    {
        Console.WriteLine("设备报警信息写入Redis异常：" + JsonConvert.SerializeObject(status) + JsonConvert.SerializeObject(e));
    }
}
void VehicleAlarmFILEDataToRedis(VehicleAlarm status, CSRedisClient redisDb3, string alarmFile)
{
    try
    {
        if (status.url.Contains("jpg"))
        {
            redisDb3.Set("alarmIdjpg" + status.alarmId, alarmFile, new TimeSpan(12, 0, 30));//
        }
        else if (status.url.Contains("mp4"))
        {
            redisDb3.Set("alarmIdmp4" + status.alarmId, alarmFile, new TimeSpan(12, 0, 30));//
        }
        else
            redisDb3.Set("alarmId" + status.alarmId, alarmFile, new TimeSpan(12, 0, 30));//
    }
    catch (Exception e)
    {
        Console.WriteLine("设备报警文件地址信息写入Redis异常：" + JsonConvert.SerializeObject(status) + JsonConvert.SerializeObject(e));
    }
}
async void Producer(string brokerList, string topicName, string text)
{
    Console.WriteLine(topicName + "数据生产");
    var config = new ProducerConfig { BootstrapServers = brokerList };

    using (var producer = new ProducerBuilder<string, string>(config).Build())
    {
        Console.WriteLine($"Producer {producer.Name} producing on topic {topicName}.");

        {
            string key = Guid.NewGuid().ToString();
            string val = text;
            try
            {
                // Note: Awaiting the asynchronous produce request below prevents flow of execution
                // from proceeding until the acknowledgement from the broker is received (at the 
                // expense of low throughput).
                var deliveryReport = await producer.ProduceAsync(
                    topicName, new Message<string, string> { Key = key, Value = val });

                Console.WriteLine($"delivered to: {deliveryReport.TopicPartitionOffset}");
            }
            catch (ProduceException<string, string> e)
            {
                Console.WriteLine($"failed to deliver message: {e.Message} [{e.Error.Code}] Topic:{topicName}---生产数据失败");
            }
        }

    }
}
async void Consumer(string brokerList, List<string> topics)
{
    CancellationTokenSource cts = new CancellationTokenSource();
    Console.CancelKeyPress += (_, e) =>
    {
        e.Cancel = true; // prevent the process from terminating.
        cts.Cancel();
    };
    Run_Consume(brokerList, topics, cts.Token);
}
/// <summary>
///     In this example
///         - offsets are automatically committed.
///         - no extra thread is created for the Poll (Consume) loop.
/// </summary>
void Run_Consume(string brokerList, List<string> topics, CancellationToken cancellationToken)
{
    CSRedisClient redisDb = null;
    if (redisDb == null)
    {
        redisDb = RedisHelp.GetInstance();
    }
    CSRedisClient redisDb1 = null;
    if (redisDb1 == null)
    {
        redisDb1 = RedisHelp.GetInstance1();
    }
    CSRedisClient redisDb2 = null;
    if (redisDb2 == null)
    {
        redisDb2 = RedisHelp.GetInstance2();
    }
    CSRedisClient redisDb3 = null;
    if (redisDb3 == null)
    {
        redisDb3 = RedisHelp.GetInstance3();
    }
    CSRedisClient redisDb4 = null;
    if (redisDb4 == null)
    {
        redisDb4 = RedisHelp.GetInstance4();
    }
    //string brokerList = "192.168.3.52:9092,192.168.3.146:9092,192.168.3.164:9092";
    var config = new ConsumerConfig
    {
        BootstrapServers = brokerList,
        GroupId = "STATUS_GPS_ALARM_File_ToRedis",
        EnableAutoOffsetStore = false,
        EnableAutoCommit = true,
        StatisticsIntervalMs = 5000,
        SessionTimeoutMs = 6000,
        AutoOffsetReset = AutoOffsetReset.Latest,
        EnablePartitionEof = true,
        // A good introduction to the CooperativeSticky assignor and incremental rebalancing:
        // https://www.confluent.io/blog/cooperative-rebalancing-in-kafka-streams-consumer-ksqldb/
        PartitionAssignmentStrategy = PartitionAssignmentStrategy.CooperativeSticky
    };

    // Note: If a key or value deserializer is not set (as is the case below), the 
    // deserializer corresponding to the appropriate type from Confluent.Kafka.Deserializers
    // will be used automatically (where available). The default deserializer for string
    // is UTF8. The default deserializer for Ignore returns null for all input data
    // (including non-null data).
    using (var consumer = new ConsumerBuilder<Ignore, string>(config)
        // Note: All handlers are called on the main .Consume thread.
        .SetErrorHandler((_, e) => Console.WriteLine($"Error: {e.Reason}"))
        .SetStatisticsHandler((_, json) =>
            //Console.WriteLine($"Statistics: {json}")
            Console.WriteLine("rd--")
            )
        .SetPartitionsAssignedHandler((c, partitions) =>
        {
            // Since a cooperative assignor (CooperativeSticky) has been configured, the
            // partition assignment is incremental (adds partitions to any existing assignment).
            //Console.WriteLine(
            //    "Partitions incrementally assigned: [" +
            //    string.Join(',', partitions.Select(p => p.Partition.Value)) +
            //    "], all: [" +
            //    string.Join(',', c.Assignment.Concat(partitions).Select(p => p.Partition.Value)) +
            //    "]");

            // Possibly manually specify start offsets by returning a list of topic/partition/offsets
            // to assign to, e.g.:
            // return partitions.Select(tp => new TopicPartitionOffset(tp, externalOffsets[tp]));
        })
        .SetPartitionsRevokedHandler((c, partitions) =>
        {
            // Since a cooperative assignor (CooperativeSticky) has been configured, the revoked
            // assignment is incremental (may remove only some partitions of the current assignment).
            var remaining = c.Assignment.Where(atp => partitions.Where(rtp => rtp.TopicPartition == atp).Count() == 0);
            //Console.WriteLine(
            //    "Partitions incrementally revoked: [" +
            //    string.Join(',', partitions.Select(p => p.Partition.Value)) +
            //    "], remaining: [" +
            //    string.Join(',', remaining.Select(p => p.Partition.Value)) +
            //    "]");
        })
        .SetPartitionsLostHandler((c, partitions) =>
        {
            // The lost partitions handler is called when the consumer detects that it has lost ownership
            // of its assignment (fallen out of the group).
            //Console.WriteLine($"Partitions were lost: [{string.Join(", ", partitions)}]");
        })
        .Build())
    {
        consumer.Subscribe(topics);

        try
        {
            while (true)
            {
                try
                {
                    var consumeResult = consumer.Consume(cancellationToken);

                    if (consumeResult.IsPartitionEOF)
                    {
                        //Console.WriteLine(
                        //    $"Reached end of topic {consumeResult.Topic}, partition {consumeResult.Partition}, offset {consumeResult.Offset}.");

                        continue;
                    }

                    //Console.WriteLine($"Received message at {consumeResult.TopicPartitionOffset}: {consumeResult.Message.Value}");
                    if (consumeResult.Topic == "DEVICE_STATUS")
                    {
                        try
                        {
                            DeviceStatus device = new DeviceStatus(); device = JsonConvert.DeserializeObject<DeviceStatus>(consumeResult.Message.Value);
                            DeviceStatusDataToRedis(device, redisDb, redisDb4);
                        }
                        catch (Exception e)
                        {
                            Console.WriteLine("数据转换失败1" + consumeResult.Message.Value + "-------" + e.Message);
                        }
                    }
                    else if (consumeResult.Topic == "VEHICLE_STATUS")
                    {
                        try
                        {
                            DeviceStatus device = new DeviceStatus(); device = JsonConvert.DeserializeObject<DeviceStatus>(consumeResult.Message.Value);
                            DeviceStatusDataToRedis(device, redisDb, redisDb4);
                            //Console.WriteLine(consumeResult.Message.Value);
                        }
                        catch (Exception e)
                        {
                            Console.WriteLine("数据转换失败2" + consumeResult.Message.Value + "-------" + e.Message);
                        }
                    }
                    else if (consumeResult.Topic == "VEHICLE_GPS")
                    {
                        try
                        {
                            VehicleGps vehicle = new VehicleGps(); vehicle = JsonConvert.DeserializeObject<VehicleGps>(consumeResult.Message.Value);
                            VehicleGpsDataToRedis(vehicle, redisDb, redisDb2);
                        }
                        catch (Exception e)
                        {
                            Console.WriteLine("数据转换失败3" + consumeResult.Message.Value + "-------" + e.Message);
                        }
                    }
                    else if (consumeResult.Topic == "VEHICLE_ALARM")
                    {
                        try
                        {
                            VehicleAlarm vehicle = new VehicleAlarm(); vehicle = JsonConvert.DeserializeObject<VehicleAlarm>(consumeResult.Message.Value);
                            VehicleAlarmDataToRedis(vehicle, redisDb3, consumeResult.Message.Value, redisDb);
                        }
                        catch (Exception e)
                        {
                            Console.WriteLine("数据转换失败4" + consumeResult.Message.Value + "-------" + e.Message);
                        }
                    }
                    else if (consumeResult.Topic == "VEHICLE_ALARM_FILE")
                    {
                        try
                        {
                            VehicleAlarm vehicle = new VehicleAlarm(); vehicle = JsonConvert.DeserializeObject<VehicleAlarm>(consumeResult.Message.Value);
                            VehicleAlarmFILEDataToRedis(vehicle, redisDb3, consumeResult.Message.Value);
                        }
                        catch (Exception e)
                        {
                            Console.WriteLine("数据转换失败5" + consumeResult.Message.Value + "-------" + e.Message);
                        }
                    }
                    else
                    {

                    }
                    try
                    {
                        // Store the offset associated with consumeResult to a local cache. Stored offsets are committed to Kafka by a background thread every AutoCommitIntervalMs. 
                        // The offset stored is actually the offset of the consumeResult + 1 since by convention, committed offsets specify the next message to consume. 
                        // If EnableAutoOffsetStore had been set to the default value true, the .NET client would automatically store offsets immediately prior to delivering messages to the application. 
                        // Explicitly storing offsets after processing gives at-least once semantics, the default behavior does not.
                        consumer.StoreOffset(consumeResult);
                    }
                    catch (KafkaException e)
                    {
                        Console.WriteLine($"Store Offset error: {e.Error.Reason}");
                    }
                }
                catch (ConsumeException e)
                {
                    Console.WriteLine($"Consume error: {e.Error.Reason}");
                }
            }
        }
        catch (OperationCanceledException)
        {
            Console.WriteLine("Closing consumer.");
            consumer.Close();
        }
    }
}